
#pragma once

#include <atomic>
#include <cmath>
#include <future>
#include <iostream>
#include <memory>
#include <sstream>
#include <thread>
#include <vector>
#include "can_client.h"



namespace canbus
{
  template <typename SensorType>
  class CanReceiver
  {
  public:
    CanReceiver() = default;
    virtual ~CanReceiver() = default;


    ErrorCode Init(CanClient *can_client,bool enable_log);

    bool IsRunning() const;
    ErrorCode Start();
    void Stop();

  private:
    void RecvThreadFunc();

    int32_t Start(bool is_blocked);

  private:
    std::atomic<bool> is_running_ = {false};
    CanClient *can_client_ = nullptr;
    bool enable_log_ = false;
    bool is_init_ = false;
    std::future<void> async_result_;
  };


  template <typename SensorType>
  ErrorCode CanReceiver<SensorType>::Init(
      CanClient *can_client,bool enable_log)
  {
    can_client_ = can_client;
    enable_log_ = enable_log;
    if (can_client_ == nullptr)
    {
      std::cout << "Invalid can client.";
      return ErrorCode::CANBUS_ERROR;
    }
    is_init_ = true;
    return ErrorCode::OK;
  }

  template <typename SensorType>
  void CanReceiver<SensorType>::RecvThreadFunc()
  {

    int32_t receive_error_count = 0;
    int32_t receive_none_count = 0;
    const int32_t ERROR_COUNT_MAX = 10;
    auto default_period = 10 * 1000;

    while (IsRunning())
    {
      std::vector<CanFrame> can_buf;
      int32_t frame_num = MAX_CAN_RECV_FRAME_LEN;
      if (can_client_->Receive(&can_buf, &frame_num) != ErrorCode::OK)
      {
        std::cout<< "Received " << receive_error_count << " error messages.";
        // cyber::USleep(default_period);
        std::this_thread::sleep_for(std::chrono::milliseconds(default_period));
        continue;
      }
      receive_error_count = 0;

      if (can_buf.size() != static_cast<size_t>(frame_num))
      {
        std::cout << "Receiver buf size [" << can_buf.size()
                          << "] does not match can_client returned length["
                          << frame_num << "].";
      }

      if (frame_num == 0)
      {
        std::cout<< "Received " << receive_none_count << " empty messages.";
        // cyber::USleep(default_period);
        std::this_thread::sleep_for(std::chrono::milliseconds(default_period));
        continue;
      }
      receive_none_count = 0;

      for (const auto &frame : can_buf)
      {
        uint8_t len = frame.len;
        uint32_t uid = frame.id;
        const uint8_t *data = frame.data;
        // pt_manager_->Parse(uid, data, len); //  解析数据
        if (enable_log_)
        {
          std::cout << "recv_can_frame#" << frame.data;
        }
      }
      //cyber::Yield();//=====================================
    }
    std::cout << "Can client receiver thread stopped.";
  }

  template <typename SensorType>
  bool CanReceiver<SensorType>::IsRunning() const
  {
    return is_running_.load();
  }

  template <typename SensorType>
  ErrorCode CanReceiver<SensorType>::Start()
  {
    if (is_init_ == false)
    {
      return ErrorCode::CANBUS_ERROR;
    }
    is_running_.exchange(true);

    async_result_ = std::async(&CanReceiver<SensorType>::RecvThreadFunc, this); // C++ 异步线程
    return ErrorCode::OK;
  }

  template <typename SensorType>
  void CanReceiver<SensorType>::Stop()
  {
    if (IsRunning())
    {
      std::cout << "Stopping can client receiver ...";
      is_running_.exchange(false);
      async_result_.wait();
    }
    else
    {
      std::cout << "Can client receiver is not running.";
    }
    std::cout << "Can client receiver stopped [ok].";
  }

} // namespace canbus
